Postgres 中的自动向量嵌入

文章背景与核心概要

Supabase 近期发布了自动向量嵌入（Automatic Embeddings）功能，这是一项能够在 Postgres 内部直接自动化生成和更新向量嵌入的强大新特性。通过将 Supabase Vector (pgvector)、队列 (pgmq)、定时任务 (pg_cron)、pg_net 以及 Edge Functions 相结合，开发人员可以彻底摆脱外部同步流水线的束缚。这种原生方法确保了数据和向量嵌入的完美同步，同时避免了管理独立后台工作线程的复杂性、数据漂移问题，以及对数据库写入操作带来的阻塞延迟。

本文详细剖析了传统外部嵌入流水线的痛点，对比了不同的架构设计模式，并重点介绍了基于触发器的异步嵌入推荐方案。通过具体的 SQL 代码示例和实现步骤，文章展示了如何在 Supabase 环境下优雅地构建自动化的向量嵌入流水线，助力开发者更高效地实现 RAG、语义搜索和推荐等高级 AI 功能。

Executive Summary

Supabase has released Automatic Embeddings, a powerful new capability that automates the generation and updating of vector embeddings directly inside Postgres. By combining Supabase Vector (pgvector), Queues (pgmq), Cron (pg_cron), pg_net, and Edge Functions, developers can eliminate external synchronization pipelines. This native approach ensures data and embeddings stay perfectly in sync without the complexity of managing separate background workers, dealing with data drift, or adding blocking latency to database writes.

Supabase 发布了自动向量嵌入（Automatic Embeddings），这是一项强大的新功能，可在 Postgres 内部直接自动生成和更新向量嵌入。通过结合 Supabase Vector (pgvector)、队列 (pgmq)、Cron (pg_cron)、pg_net 和 Edge Functions，开发人员可以消除外部同步流水线。这种原生方法确保数据和嵌入保持完美同步，而无需管理单独的后台工作线程、处理数据漂移或向数据库写入添加阻塞延迟的复杂性。

The Problem: External Embedding Pipelines

Implementing semantic features (like RAG, semantic search, and recommendations) traditionally requires building and maintaining an external synchronization pipeline:

1. Store source content (documents, tickets, articles).
2. Generate an embedding via an external model API.
3. Store the vector result in the database.
4. Re-run jobs whenever content changes.
5. Handle retries for timeouts and model failures.

Common Failure Points

• Data Drift: Forgetting to re-embed updated content degrades search quality over time.
• Latency: Synchronous API calls on the write path add unnecessary delay.
• Lack of Resilience: Background worker crashes or queue failures can go unnoticed until features break.
• Schema Duplication: Application code ends up duplicating logic that belongs in the database schema.

问题：外部嵌入流水线

实现语义功能（如 RAG、语义搜索和推荐）传统上需要构建和维护外部同步流水线：

1. 存储源内容（文档、工单、文章）。
2. 通过外部模型 API 生成嵌入。
3. 将向量结果存储在数据库中。
4. 每当内容更改时重新运行任务。
5. 处理超时和模型故障的重试。

常见故障点

• 数据漂移： 忘记对更新的内容重新嵌入，会导致搜索质量随时间下降。
• 延迟： 写入路径上的同步 API 调用会增加不必要的延迟。
• 缺乏弹性： 后台工作线程崩溃或队列故障可能在功能损坏之前一直不被察觉。
• 架构重复： 应用程序代码最终重复了本应属于数据库架构的逻辑。

What Are Automatic Embeddings?

Automatic embeddings shift the coordination of vector generation directly into Postgres. While inference still happens via an external model, the lifecycle management is handled via native database primitives.

When a row is inserted or updated, Postgres automatically enqueues a transactional background job. This job runs asynchronously, handles retries, and writes the resulting vector back to the database.

Key Benefits

• No Drift: Embeddings update automatically alongside source content.
• Bring Your Own Model: Compatible with any API endpoint that returns a vector.
• Pure SQL Interface: Enqueue, inspect, and retry embedding jobs entirely within SQL.

什么是自动向量嵌入？

自动嵌入将向量生成的协调工作直接转移到了 Postgres 内部。虽然推理仍然通过外部模型进行，但生命周期管理是通过本地数据库原语处理的。

当插入或更新行时，Postgres 会自动将事务性后台任务加入队列。该任务异步运行，处理重试，并将生成的向量写回数据库。

核心优势

• 无数据漂移： 嵌入与源内容一起自动更新。
• 自带模型（BYOM）： 兼容任何返回向量的 API 端点。
• 纯 SQL 接口： 完全在 SQL 中对嵌入任务进行入队、检查和重试。

Design Patterns for Generating Embeddings

1. Generated Columns

create table documents (
  id uuid primary key,
  content text,
  embedding vector(1536) generated always as (embed(content)) stored
);

2. Trigger-Based Asynchronous Embeddings (Recommended)

The Supabase approach leverages a robust stack of Postgres extensions: * SQL Triggers: Detect inserts and updates. * pgmq: Enqueues embedding jobs inside a transactional message queue. * pg_net: Sends asynchronous HTTP requests to Edge Functions (and downstream embedding providers like OpenAI). * pg_cron: Runs background workers to process the queue reliably. * pgvector: Stores and indexes the resulting embeddings.

生成嵌入的设计模式

1. 生成列 (Generated Columns)

create table documents (
  id uuid primary key,
  content text,
  embedding vector(1536) generated always as (embed(content)) stored
);

* 优点： 声明式且自动化。 * 缺点： 仅对本地快速模型实用。在写入时通过 embed() 利用外部 API 会阻塞写入路径且无法扩展。

2. 基于触发器的异步嵌入（推荐）

Supabase 的方法利用了一套强大的 Postgres 扩展栈： * SQL 触发器： 检测插入和更新。 * pgmq： 在事务性消息队列中对嵌入任务进行入队。 * pg_net： 向 Edge Functions（以及下游嵌入提供商如 OpenAI）发送异步 HTTP 请求。 * pg_cron： 运行后台工作线程以可靠地处理队列。 * pgvector： 存储和索引生成的嵌入。

How to Use Automatic Embeddings

1. Set Up the Table

Create a documents table equipped with an indexing strategy for vector search:

-- Table to store documents with embeddings
create table documents (
  id integer primary key generated always as identity,
  title text not null,
  content text not null,
  embedding halfvec(1536),
  created_at timestamp with time zone default now()
);

-- Index for vector search over document embeddings
create index on documents using hnsw (embedding halfvec_cosine_ops);

2. Create the Embedding Pipeline

Define an embedding_input function to configure how source columns are concatenated for the embedding model:

-- Customize the input for embedding generation
-- e.g. Concatenate title and content with a markdown header
create or replace function embedding_input(doc documents)
returns text
language plpgsql
immutable
as $$
begin
  return '# ' || doc.title || E'\n\n' || doc.content;
end;
$$;

Next, add triggers for insert and update events to queue jobs asynchronously:

-- Trigger for insert events
create trigger embed_documents_on_insert
  after insert
  on documents
  for each row
  execute function util.queue_embeddings('embedding_input', 'embedding');

-- Trigger for update events
create trigger embed_documents_on_update
  after update of title, content -- must match the columns in embedding_input()
  on documents
  for each row
  execute function util.queue_embeddings('embedding_input', 'embedding');

The background Edge Function coordinates with your chosen inference provider (e.g., OpenAI):

/**
 * Generates an embedding for the given text.
 */
async function generateEmbedding(text: string) {
  const response = await openai.embeddings.create({
    model: 'text-embedding-3-small',
    input: text,
  })
  const [data] = response.data
  if (!data) {
    throw new Error('failed to generate embedding')
  }
  return data.embedding
}

3. Insert and Query Data

Insert a new record into your table:

insert into documents (title, content)
values
  ('Understanding Vector Databases', 'Vector databases are specialized...');

If queried immediately, the embedding column will be temporarily null while the Edge Function processes the queue in the background. Within seconds, the background worker completes inference, populates the column, and ensures your data remains fully synchronized for vector search.

如何使用自动向量嵌入

1. 设置表结构

创建一个配备向量搜索索引策略的 documents 表：

-- Table to store documents with embeddings
create table documents (
  id integer primary key generated always as identity,
  title text not null,
  content text not null,
  embedding halfvec(1536),
  created_at timestamp with time zone default now()
);

-- Index for vector search over document embeddings
create index on documents using hnsw (embedding halfvec_cosine_ops);

2. 创建嵌入流水线

定义一个 embedding_input 函数来配置如何为嵌入模型拼接源列：

-- Customize the input for embedding generation
-- e.g. Concatenate title and content with a markdown header
create or replace function embedding_input(doc documents)
returns text
language plpgsql
immutable
as $$
begin
  return '# ' || doc.title || E'\n\n' || doc.content;
end;
$$;

接下来，添加用于插入和更新事件的触发器，以异步方式将任务排队：

-- Trigger for insert events
create trigger embed_documents_on_insert
  after insert
  on documents
  for each row
  execute function util.queue_embeddings('embedding_input', 'embedding');

-- Trigger for update events
create trigger embed_documents_on_update
  after update of title, content -- must match the columns in embedding_input()
  on documents
  for each row
  execute function util.queue_embeddings('embedding_input', 'embedding');

后台 Edge Function 与您选择的推理提供商（例如 OpenAI）进行协调：

/**
 * Generates an embedding for the given text.
 */
async function generateEmbedding(text: string) {
  const response = await openai.embeddings.create({
    model: 'text-embedding-3-small',
    input: text,
  })
  const [data] = response.data
  if (!data) {
    throw new Error('failed to generate embedding')
  }
  return data.embedding
}

3. 插入和查询数据

向表中插入一条新记录：

insert into documents (title, content)
values
  ('Understanding Vector Databases', 'Vector databases are specialized...');

如果立即进行查询，embedding 列将暂时为 null，因为 Edge Function 正在后台处理队列。几秒钟内，后台工作线程将完成推理、填充该列，并确保您的数据完全同步，可用于向量搜索。

Conclusion

Get started with automatic embeddings today: * Read the full implementation details in the official docs. * Sign in to Supabase to try it out in your project.

结论

立即开始使用自动向量嵌入： * 在官方文档中阅读完整的实现细节。 * 登录 Supabase并在您的项目中进行试用。